package mylab.starters.eventbus.support.disruptor;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.Getter;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import mylab.starters.eventbus.api.EventListener;
import mylab.starters.eventbus.api.EventListenerRegistry;
import mylab.starters.eventbus.api.EventModel;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

import jakarta.annotation.PostConstruct;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;


@Slf4j
@RequiredArgsConstructor
@Component("disruptor")
@Scope("prototype") // 线程安全问题
public class DisruptorEventListenerRegistry implements EventListenerRegistry<EventModel>, AutoCloseable {

    /**
     * disruptor事件处理器
     */
    @Getter
    @Setter
    private Disruptor<EventModel> disruptor;

    @NonNull
    final List<EventListener> eventListeners;


    /**
     * RingBuffer的大小
     */
    private final int DEFAULT_RING_SIZE = 1024 * 1024;

    /**
     * 事件工厂
     */
    private EventFactory<EventModel> eventFactory = new EventModelFactory();

    @Override
    public void initRegistryEventListener(List<EventListener> eventConsumerList) {

        disruptor = new Disruptor<>(eventFactory, DEFAULT_RING_SIZE, createThreadFactory(), ProducerType.SINGLE, new BlockingWaitStrategy());
        EventHandler[] dataListener = eventConsumerList.stream().map(param -> {
            EventListener<EventModel> eventModelEventListener = param;
            return eventModelEventListener;
        }).collect(Collectors.toList()).toArray(new EventHandler[eventConsumerList.size()]);
        log.info("注册服务信息接口：{}", dataListener);

        disruptor.handleEventsWith(dataListener);
        disruptor.start();
    }

    @Override
    public void publish(EventModel param) {
        publishEvent(param);
    }

    public void publishEvent(EventModel... eventModels) {
        Objects.requireNonNull(disruptor, "当前disruptor核心控制器不可以为null");
        Objects.requireNonNull(eventModels, "当前eventModels事件控制器不可以为null");

        // 发布事件
        final RingBuffer<EventModel> ringBuffer = disruptor.getRingBuffer();
        try {
            final List<EventModel> dataList = Arrays.stream(eventModels).collect(Collectors.toList());

            for (EventModel element : dataList) {

                // 请求下一个序号
                long sequence = ringBuffer.next();

                // 获取该序号对应的事件对象
                EventModel event = ringBuffer.get(sequence);
                event.setTopic(element.getTopic());
                event.setEntity(element.getEntity());
                ringBuffer.publish(sequence);

            }
        } catch (Exception e) {
            log.error("error", e);
        }

    }


    /**
     * 关闭处理机制
     *
     * @throws Exception
     */
    @Override
    public void close() throws Exception {
        if (Objects.nonNull(disruptor)) disruptor.shutdown();

    }

    @PostConstruct
    public void init() {
        log.info("开始初始化Disruptor事件监听器的组件服务");
        initRegistryEventListener(eventListeners);
        log.info("完成初始化Disruptor事件监听器的组件服务");
    }

    private static ThreadFactory createThreadFactory() {
        AtomicInteger integer = new AtomicInteger();

        return r -> new Thread(r, "disruptor-" + integer.incrementAndGet());
    }


}
